package org.example;

import org.apache.flink.api.common.ExecutionMode;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.example.bean.WaterSensor;
import org.example.function.FilterFunctionImpl;

/*
 * @Description: dataStream有界流,lambda表达式，泛型擦除，显示指定返回类型
 */
public class WordCountExecutionDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//        DataGeneratorSource<String> dataGeneratorSource =
//                new DataGeneratorSource<>(
//                        new GeneratorFunction<Long, String>() {
//                            @Override
//                            public String map(Long value) throws Exception {
//                                return "Number:"+value;
//                            }
//                        },
//                        Long.MAX_VALUE,
//                        RateLimiterStrategy.perSecond(10),
//                        Types.STRING
//                );
//        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(),"genSource")
        env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_2", 1000L, 10),
                        new WaterSensor("sensor_3", 10000L, 10),
                        new WaterSensor("sensor_3", 1000L, 10),
                        new WaterSensor("a",1L,1))
                .filter(new FilterFunctionImpl("a"))
//                .keyBy(new KeySelector<WaterSensor, String>() {
//                    @Override
//                    public String getKey(WaterSensor value) throws Exception {
//                        return value.id;
//                    }
//                })
//                        .reduce(new ReduceFunction<WaterSensor>() {
//                            @Override
//                            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
//                                System.out.println("value1" + value1);
//                                System.out.println("value2" + value2);
//                                return new WaterSensor(value1.id,value1.ts,value1.vc+value2.vc);
//                            }
//                        })
//    .map(x->Tuple2.of(x.id,x.vc)).returns(new TypeHint<Tuple2<String, Integer>>() {})
                .print();
        env.execute();
    }
    public static class UserMap implements MapFunction<WaterSensor, String> {
        @Override
        public String map(WaterSensor e) throws Exception {
            return e.id;
        }
    }

    public static class UserFilters implements FilterFunction<WaterSensor> {

        @Override
        public boolean filter(WaterSensor value) throws Exception {
            return value.id.startsWith("sensor");
        }
    }
}

